package com.tourscool.springboot.starter.redisson.queue;

import com.tourscool.springboot.starter.redisson.RedisExecBuilder;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RLock;
import org.redisson.api.RQueue;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 我长得帅，不需要注释
 *
 * @author Carlton
 * @date 2018/4/4
 */
@Component
public class DelayedQueueClient implements InitializingBean {
    @Autowired(required = false)
    private List<DelayQueueJob> mDelayQueueJobs;
    @Autowired
    private DelayQueueHandler mDelayQueueHandler;

    /**
     * 单例存储延时队列
     */
    private volatile static ConcurrentHashMap<String , RDelayedQueue<Object>> queuesMap;

    private static ConcurrentHashMap<String , RDelayedQueue<Object>> getQueuesMap() {
        if (queuesMap == null) {
            synchronized (DelayedQueueClient.class) {
                if (queuesMap == null) {
                    queuesMap = new ConcurrentHashMap<>();
                }
            }
        }
        return queuesMap;
    }

    public void send(String queueName, Object msg, long delay, TimeUnit timeUnit) {
        RedisExecBuilder.build().execNonReturn(redissonClient -> {
            final RQueue<Object> queue = redissonClient.getBlockingQueue(queueName);
            RLock lock = null;
            try {
                lock = redissonClient.getLock(queueName + this.getClass().getName());
                lock.lock();
                RDelayedQueue<Object> delayedQueue;
                if(getQueuesMap().get(queueName) == null){
                    delayedQueue = redissonClient.getDelayedQueue(queue);
                    queuesMap.put(queueName, delayedQueue);
                }else {
                    // 从单例map中获取delayedQueue
                    delayedQueue = getQueuesMap().get(queueName);
                }
                delayedQueue.offer(msg, delay, timeUnit);
            } finally {
                if (lock != null && lock.isLocked()) {
                    lock.unlock();
                }
            }
        });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (mDelayQueueJobs == null) {
            return;
        }
        for (DelayQueueJob job : mDelayQueueJobs) {
            mDelayQueueHandler.handler(job);
        }
    }
}
